package com.wjk.kylin.rpc.core.invoker.reference;

import com.wjk.kylin.rpc.core.invoker.call.CallType;
import com.wjk.kylin.rpc.core.invoker.route.LoadBalance;
import com.wjk.kylin.rpc.core.netty.AbstractClient;
import com.wjk.kylin.rpc.core.netty.client.NettyClient;
import com.wjk.kylin.rpc.core.invoker.RpcConsumerFactory;
import com.wjk.kylin.rpc.core.invoker.future.RpcFutureResponse;
import com.wjk.kylin.rpc.core.netty.params.RpcRequest;
import com.wjk.kylin.rpc.core.netty.params.RpcResponse;
import com.wjk.kylin.rpc.core.provider.RpcProviderFactory;
import com.wjk.kylin.rpc.core.serialize.Serializer;
import com.wjk.kylin.rpc.core.util.RpcException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.Proxy;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * rpc reference bean, use by api
 *
 * @author wjk
 */
public class RpcReferenceBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(RpcReferenceBean.class);

    private Serializer.Algorithm serializer;
    private Class<?> interfaceName;
    private String version;
    private long timeout;
    private String address;
    private String accessToken;
    private RpcConsumerFactory consumerFactory;
    private LoadBalance.Route loadBalance;
    private CallType callType;

    public void setCallType(CallType callType) {
        this.callType = callType;
    }

    public void setLoadBalance(LoadBalance.Route loadBalance) {
        this.loadBalance = loadBalance;
    }

    public void setSerializer(Serializer.Algorithm serializer) {
        this.serializer = serializer;
    }

    public void setInterfaceName(Class<?> interfaceName) {
        this.interfaceName = interfaceName;
    }

    public void setVersion(String version) {
        this.version = version;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public void setAccessToken(String accessToken) {
        this.accessToken = accessToken;
    }

    public void setConsumerFactory(RpcConsumerFactory consumerFactory) {
        this.consumerFactory = consumerFactory;
    }


    public Serializer.Algorithm getSerializer() {
        return serializer;
    }

    public long getTimeout() {
        return timeout;
    }

    public RpcConsumerFactory getConsumerFactory() {
        return consumerFactory;
    }

    public Class<?> getInterfaceName() {
        return interfaceName;
    }


    private final AbstractClient client = new NettyClient();


    public void initClient() throws Exception {

        if (this.serializer == null) {
            throw new RpcException("kylin-rpc reference serializer missing.");
        }
        if (this.interfaceName == null) {
            throw new RpcException("kylin-rpc reference interfaceName missing.");
        }
        if (this.timeout < 0) {
            this.timeout = -1;
        }
        if (this.consumerFactory == null) {
            throw new RpcException("kylin-rpc reference consumerFactory missing.");
        }
        if (this.loadBalance == null) {
            this.loadBalance = LoadBalance.Route.Round;
        }
        // init Client
        client.init(this);
    }


    // ---------------------- util ----------------------

    public Object getObject() throws Exception {

        // initClient
        initClient();

        // newProxyInstance 重写 动态代理 ，每次调用返回都会走这个方法
        return Proxy.newProxyInstance(Thread.currentThread()
                        .getContextClassLoader(), new Class[]{interfaceName},
                (proxy, method, args) -> {

                    // method param
                    String className = method.getDeclaringClass().getName();    // interfaceName.getName()
                    String isVersion = version;
                    String methodName = method.getName();
                    Class<?>[] parameterTypes = method.getParameterTypes();

                    // filter method like "Object.toString()"
                    if (className.equals(Object.class.getName())) {
                        LOGGER.error(">>>>>>>>>>> kylin-rpc proxy class-method not support [{}#{}]", className, methodName);
                        throw new RpcException("kylin-rpc proxy class-method not support");
                    }

                    // address
                    String finalAddress = address;
                    if (StringUtils.isBlank(finalAddress)) {
                        if (consumerFactory != null && consumerFactory.getRegister() != null) {
                            // discovery
                            String serviceKey = RpcProviderFactory.makeServiceKey(className, isVersion);
                            TreeSet<String> addressSet = consumerFactory.getRegister().discovery(serviceKey);
                            if (!CollectionUtils.isEmpty(addressSet)) {
                                if (addressSet.size() == 1) {
                                    finalAddress = addressSet.first();
                                } else {
                                    finalAddress = loadBalance.route(serviceKey, addressSet);
                                }
                            }
                        }
                    }
                    if (StringUtils.isBlank(finalAddress)) {
                        throw new RpcException("kylin-rpc reference bean[" + className + "] address empty");
                    }

                    // request
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setCreateMillisTime(System.currentTimeMillis());
                    rpcRequest.setAccessToken(accessToken);
                    rpcRequest.setClassName(className);
                    rpcRequest.setMethodName(methodName);
                    rpcRequest.setParameterTypes(parameterTypes);
                    rpcRequest.setParameters(args);
                    rpcRequest.setVersion(version);

                    // send
                    if (callType == null || CallType.SYNC == callType) {
                        RpcFutureResponse futureResponse = new RpcFutureResponse(consumerFactory, rpcRequest);
                        try {
                            LOGGER.debug(">>>>>>>>>>> kylin-rpc invoke response request:{}", rpcRequest);

                            // do invoke
                            client.asyncSend(finalAddress, rpcRequest);

                            // future get
                            RpcResponse rpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);//
                            LOGGER.debug(">>>>>>>>>>> kylin-rpc invoke response result:{}", rpcResponse);

                            if (rpcResponse.getErrorMsg() != null) {
                                throw new RpcException(rpcResponse.getErrorMsg());
                            }
                            return rpcResponse.getResult();
                        } catch (Exception e) {
                            LOGGER.error(">>>>>>>>>>> kylin-rpc invoke error, address:{}, request:{}", finalAddress, rpcRequest);

                            throw (e instanceof RpcException) ? e : new RpcException(e);
                        } finally {
                            // future-response remove
                            futureResponse.removeConsumerFuture();
                        }
                    } else if (CallType.FUTURE == callType) {
//                        // future-response set
//                        RpcFutureResponse futureResponse = new RpcFutureResponse(consumerFactory, rpcRequest);
//                        try {
//                            // invoke future set
////                            XxlRpcInvokeFuture invokeFuture = new XxlRpcInvokeFuture(futureResponse);
////                            XxlRpcInvokeFuture.setFuture(invokeFuture);
//
//                            // do invoke
//                            client.asyncSend(finalAddress, rpcRequest);
//
//                        } catch (Exception e) {
//                            LOGGER.info(">>>>>>>>>>> kylin-rpc invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
//                            // future-response remove
//                            futureResponse.removeConsumerFuture();
//
//                            throw (e instanceof RpcException) ? e : new RpcException(e);
//                        }
                        return null;
                    } else if (CallType.CALLBACK == callType) {
//
//                        // get callback
//                        XxlRpcInvokeCallback finalInvokeCallback = invokeCallback;
//                        XxlRpcInvokeCallback threadInvokeCallback = XxlRpcInvokeCallback.getCallback();
//                        if (threadInvokeCallback != null) {
//                            finalInvokeCallback = threadInvokeCallback;
//                        }
//                        if (finalInvokeCallback == null) {
//                            throw new RpcException("kylin-rpc RpcInvokeCallback（CallType=" + CallType.CALLBACK.name() + "） cannot be null.");
//                        }
//
//                        // future-response set
//                        RpcFutureResponse futureResponse = new RpcFutureResponse(consumerFactory, rpcRequest, finalInvokeCallback);
//                        try {
//                            client.asyncSend(finalAddress, rpcRequest);
//                        } catch (Exception e) {
//                            LOGGER.info(">>>>>>>>>>> kylin-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, rpcRequest);
//
//                            // future-response remove
//                            futureResponse.removeConsumerFuture();
//
//                            throw (e instanceof RpcException) ? e : new RpcException(e);
//                        }

                        return null;
                    } else if (CallType.ONE == callType) {
                        client.asyncSend(finalAddress, rpcRequest);
                        return null;
                    } else {
                        throw new RpcException("kylin-rpc callType[" + callType + "] invalid");
                    }

                });
    }

}
